package com.zhss.dfs.namenode.server.editlog;

import com.alibaba.fastjson.JSON;
import com.zhss.dfs.namenode.server.context.DfsEnv;
import com.zhss.dfs.namenode.server.context.Environment;

import java.io.*;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;


public class EditLogBuffer {
    private Environment env = DfsEnv.getInstance();
    volatile ByteArrayOutputStream buffer = new ByteArrayOutputStream(env.getEditlogBufferLimit() * 2);
    public final String LogDirPath = env.getEditlogDirPath();
    public Long endTxid;

    public void write(EditLog editLog) throws IOException {
        endTxid = editLog.getTxid();
        buffer.write(JSON.toJSONString(editLog).getBytes());
        buffer.write("\n".getBytes());
//        System.out.println("当前缓冲区大小为:" + buffer.size());

    }

    /**
     * 将buffer中的数据刷入磁盘中
     */
    public Long flush(Long startTxid) throws IOException {
        byte[] data = buffer.toByteArray();
        ByteBuffer byteBuffer = ByteBuffer.wrap(data);

        FileChannel channel = null;
        FileOutputStream fileOutputStream = null;
        RandomAccessFile file = null;
        try {
            File path = new File(LogDirPath);
            if (!path.exists()) {
                path.createNewFile();
            }
            String filename = startTxid + "-" + endTxid;
            file = new RandomAccessFile(LogDirPath + "/editlog/" + filename + ".log", "rw");
            DoubleBuffer.flushedTxids.add(filename);
            fileOutputStream = new FileOutputStream(file.getFD());
            channel = fileOutputStream.getChannel();
            channel.position(channel.size());
            channel.write(byteBuffer);
            channel.force(false); //强制把数据刷入磁盘
            return endTxid + 1;
        } catch (Exception e) {
            e.printStackTrace();
        } finally {
            if (fileOutputStream != null) {
                fileOutputStream.close();
            }
            if (file != null) {
                file.close();
            }
            if (channel != null) {
                channel.close();
            }
        }

        return 0l;
    }

    public void clear() {
        buffer.reset();
    }

    public int size() {
        return buffer.size();
    }

    public byte[] getBufferData() {
        return buffer.toByteArray();
    }

}
